// -*-mode:c++; coding:utf-8-*-

#include "logger.hpp"

#include "multi_row_mutator.hpp"

namespace xbase
{
	void multi_row_mutator::async_apply(const std::string &table_name,
					    const container_ptr &mutations,
					    uint64_t timestamp,
					    handler_type handler)
	{
		if(mutations->empty())
		{
			this->get_upper_level_service().
				dispatch(boost::bind(handler, 0, error_code()));
			return;
		}

		m_table_name = table_name;
		m_mutations = mutations;
		m_timestamp = timestamp;
		m_handler = handler;

		m_success_row = 0;
		m_start_pos = mutations->begin();
		m_tablet_info_invalid = false;
		m_retry_times = 0;

		m_meta_data.meta_got_time = 0;

		this->get_lower_level_service().dispatch(
			boost::bind(&this_type::on_async_apply, this_ptr())
			);
	}

	void multi_row_mutator::async_apply(const std::string &table_name,
					    const container_ptr &mutations,
					    handler_type handler)
	{
		async_apply(table_name,
			    mutations,
			    cur_micro_seconds(),
			    handler);
	}

	void multi_row_mutator::on_async_apply()
	{
		const std::string &first_row_name = (*m_start_pos)->row_name();
		m_location_cache.async_get_meta(
			boost::bind(&this_type::on_got_tablet,
				    this_ptr(),
				    _1, _2),
			m_table_name,
			first_row_name,
			m_tablet_info_invalid,
			m_meta_data.meta_got_time);
	}

	void multi_row_mutator::on_got_tablet(location_cache_value meta, const error_code &ec)
	{
		if (ec)
		{
			LOG(DEBUG, "got tablet failed");
			if((ec == error::tablet_not_exists) ||
			   (ec == error::table_not_exists))
			{
				on_retry(0, ec);
			}
			else
			{
				on_error(0, ec);
			}
		}
		else
		{
			m_tablet_info_invalid = false;
			
			m_meta_data = meta;

			container_iterator new_end;
			if(m_meta_data.tablet_info_p->m_end_row.empty())
			{
				new_end = m_mutations->end();
			}
			else
			{
				const container_iterator end = m_mutations->end();
				new_end = std::lower_bound(m_start_pos, end,
							   m_meta_data.tablet_info_p->m_end_row,
							   comp());
			}
			// 结束行变化了，也必须重置
			if((m_req_ptr) &&
			   (m_end_pos != new_end))
				m_req_ptr.reset();
			m_end_pos = new_end;

			m_session_manager.async_connect(m_meta_data.server_addr_p->to_endpoint(),
							boost::bind(&this_type::on_session_connected,
								    this_ptr(),
								    _1, _2));
		}
	}

	void multi_row_mutator::on_session_connected(session_ptr session,
						     const error_code &ec)
	{
		if(ec)
		{
			LOG(DEBUG, "session connect failed");
			on_retry(0, ec);
			return;
		}

		if(! m_req_ptr) // request changed
		{
			m_req_ptr.reset(new batch_write_req());
			protocol::batch_write_request_v2 &body = m_req_ptr->body();
			body.m_tablet_name = m_meta_data.tablet_info_p->tablet_name();
			for(container_iterator it = m_start_pos; it != m_end_pos; ++ it)
			{
				body.m_rows.push_back(protocol::row_data_v2(*it, m_timestamp));
				//(*it)->fill_record_vector(body.m_rows.back().m_records, m_timestamp);
			}
		}
		if(! m_tablet_mutator)
		{
			m_tablet_mutator.reset(new_async_module(multi_row_mutator_on_tablet, m_io_service_pool));
		}
		m_tablet_mutator->async_apply(session,
					      m_req_ptr,
					      boost::bind(&this_type::on_complete,
							  this_ptr(),
							  _1, _2));
	}

	void multi_row_mutator::on_complete(std::size_t success_row, const error_code &ec)
	{
		if(ec)
		{
			LOG(ERROR, "async apply on tablet mutator failed");
			if((ec == error::tablet_not_exists) ||
			   (ec == error::table_not_exists) ||
			   (ec == error::row_not_in_range))
			{
				on_retry(success_row, ec);
			}
			else
			{
				on_error(success_row, ec);
			}
		}
		else
		{
			m_start_pos = m_end_pos;
			if(m_start_pos != m_mutations->end())
			{
				// 还没有写完，接下来应该是切换子表，所以将以下变量重置
				m_retry_times = 0;
				m_tablet_info_invalid = false;
				m_meta_data.meta_got_time = 0;
				m_req_ptr.reset();

				m_success_row += success_row;

				on_async_apply();
			}
			else
			{
				dispatch(success_row, ec);
			}
		}
	}

	void multi_row_mutator::on_retry(std::size_t success_row, const error_code &ec)
	{
		if(m_options.can_retry(m_retry_times))
		{
			const std::string &first_row_name = (*m_start_pos)->row_name();
			LOG(WARN, "multi_row_mutator fail, table_name = " << m_table_name
			    << ", first mutation row = " << dump_string_char_with_hex(first_row_name)
			    << ", mutation row number = " << m_mutations->size()
			    << ", timestamp = " << m_timestamp
			    << ", already success_row = " << m_success_row
			    << ", tablet info = "
			    << (m_meta_data.tablet_info_p ? m_meta_data.tablet_info_p->to_string() : "NULL")
			    << ", server address = "
			    << (m_meta_data.server_addr_p ? m_meta_data.server_addr_p->to_string() : "NULL")
				<< ", meta_got_time = " << m_meta_data.meta_got_time
			    << ", tablet_info_invalid = " << std::boolalpha << m_tablet_info_invalid
			    << ", cur success_row = " << success_row
			    << ", retry_times = " << m_retry_times
			    << ", error = " << ec << " - " << ec.message());

			// skip success rows
			std::advance(m_start_pos, success_row);

			if(m_start_pos == m_mutations->end()) // no data, success
			{
				dispatch(success_row, error_code());
				return;
			}

			// 起始行变化了，则行请求也必须重置
			if(success_row != 0)
				m_req_ptr.reset();

			m_success_row += success_row;
			++ m_retry_times;
			m_tablet_info_invalid = true;

			if(! m_timer)
				m_timer.reset(new boost::asio::deadline_timer(this->get_lower_level_service()));
			m_timer->expires_from_now(m_options.get_retry_interval());
			m_timer->async_wait(boost::bind(&this_type::on_async_apply, this_ptr()));
		}
		else
		{
			on_error(success_row, ec);
		}
	}

	void multi_row_mutator::on_error(std::size_t success_row, const error_code &ec)
	{
		const std::string &first_row_name = (*m_start_pos)->row_name();
		LOG(ERROR, "multi_row_mutator error, table_name = " << m_table_name
		    << ", first mutation row = " << dump_string_char_with_hex(first_row_name)
		    << ", mutation row number = " << m_mutations->size()
		    << ", timestamp = " << m_timestamp
		    << ", already success_row = " << m_success_row
		    << ", tablet info = "
		    << (m_meta_data.tablet_info_p ? m_meta_data.tablet_info_p->to_string() : "NULL")
		    << ", server address = "
		    << (m_meta_data.server_addr_p ? m_meta_data.server_addr_p->to_string() : "NULL")
			<< ", meta_got_time = " << m_meta_data.meta_got_time
		    << ", tablet_info_invalid = " << std::boolalpha << m_tablet_info_invalid
		    << ", cur success_row = " << success_row
		    << ", retry_times = " << m_retry_times
		    << ", error = " << ec << " - " << ec.message());

		dispatch(success_row, ec);
	}

	void multi_row_mutator::dispatch(std::size_t success_row, const error_code &ec)
	{
		const std::size_t total_success_row = success_row + m_success_row;

		// release resouces
		m_mutations.reset();
		m_req_ptr.reset();
		// m_tablet_mutator 可以重用
		// 子表信息保留

		handler_type h = m_handler;
		m_handler = handler_type();

		assert(h);
		this->get_upper_level_service().
			dispatch(boost::bind(h, total_success_row, ec));
	}
}
